package com.mrx.commons.logback.collector;

import com.alibaba.fastjson2.JSON;
import com.mrx.commons.logback.collector.dto.OpenObserveResponse;
import com.mrx.commons.logback.config.XLogCollectorProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * openObserve 日志收集器
 *
 * @author Mr.X
 * @since 2023-12-09 下午 8:54:50
 */
public class OpenObserveLogCollector {

    private static final String prefix = "x-log-collector";
    private static final HttpClient client = HttpClient.newBuilder().build();
    private static final ExecutorService pool = Executors.newSingleThreadExecutor(new CustomizableThreadFactory(prefix));
    private static final Logger logger = LoggerFactory.getLogger(OpenObserveLogCollector.class);
    private static final AtomicReference<XLogCollectorProperties> properties = new AtomicReference<>();
    private static volatile boolean serverError = false;
    private static volatile boolean checkedServer = false;
    private static final int CHECK_SERVER_TIME_OUT = 1;

    public static void setProperties(XLogCollectorProperties properties) {
        OpenObserveLogCollector.properties.set(properties);
    }

    @SuppressWarnings("StatementWithEmptyBody")
    public static void submitToOpenObserve(byte[] content) {
        if (serverError) return;
        Future<?> submitted = pool.submit(() -> {
            while (properties.compareAndSet(null, properties.get())) {
                // cas
            }
            sendToOpenObserve(content);
        });
        // 首次日志消息变更检测下服务器是否可用
        if (!checkedServer) {
            try {
                submitted.get(CHECK_SERVER_TIME_OUT, TimeUnit.SECONDS);
            } catch (Exception e) {
                logger.info("o2 日志服务器不可用");
                checkedServer = true;
                serverError = true;
            }
        }
    }

    public static void sendToOpenObserve(byte[] content) {
        sendToOpenObserve(content, properties.get().getStream());
    }

    @SuppressWarnings("unused")
    public static void sendToOpenObserve(String content) {
        sendToOpenObserve(content, properties.get().getStream());
    }

    public static void sendToOpenObserve(byte[] content, String stream) {
        sendToOpenObserve(HttpRequest.BodyPublishers.ofByteArray(content), stream);
    }

    public static void sendToOpenObserve(String content, String stream) {
        sendToOpenObserve(HttpRequest.BodyPublishers.ofString(content), stream);
    }

    public static void sendToOpenObserve(HttpRequest.BodyPublisher body, String stream) {
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(String.format(properties.get().getServer() + "/api/%s/%s/_json", properties.get().getOrg(), stream)))
                .POST(body)
                .setHeader("Content-Type", "application/json")
                .setHeader("Authorization", "Basic " + properties.get().getToken())
                .build();
        try {
            HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
            OpenObserveResponse o2Response = JSON.parseObject(response.body(), OpenObserveResponse.class);
            boolean codeError = o2Response.getCode() != 200;
            Boolean hasFailed = Optional.ofNullable(o2Response.getStatus())
                    .map(it -> it.stream().anyMatch(item -> item.getFailed() > 0)).orElse(false);
            if (codeError || hasFailed) {
                throw new RuntimeException(JSON.toJSONString(o2Response));
            }
            logger.trace("发送日志变更消息到 o2, 结果: {}", o2Response);
        } catch (Exception e) {
            serverError = true;
            logger.error("发送日志变更消息到 o2 失败", e);
        }
    }

}
